package com.cjfather.redis.steam.controller;

import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import com.cjfather.redis.steam.annotation.constants.RedisConstants;
import com.cjfather.redis.steam.config.RedisCacheUtil;
import com.cjfather.redis.steam.service.CommonService;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import lombok.extern.slf4j.Slf4j;
/**
 * 
 * @author: todd
 * @date:   2023-08-08
 */

@Slf4j
@RestController
public class MessageController {

    @Resource
    private StringRedisTemplate stringRedisTemplate;
    
    @Resource
    private RedisCacheUtil redisCacheUtil;
    
    @Resource
    private CommonService commonService;


    private StreamOperations<String, String, String> stream = null;
    
    static AtomicInteger index = new AtomicInteger(1);
    static int processors = Runtime.getRuntime().availableProcessors() * 5;
    static ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(), r -> {
        Thread thread = new Thread(r);
        thread.setName("todd-" + index.getAndIncrement());
        thread.setDaemon(true);
        return thread;
    });

    @PostConstruct
    public void init(){
        LettuceConnectionFactory connectionFactory = (LettuceConnectionFactory) stringRedisTemplate.getConnectionFactory();
        // 不启用共享连接
        connectionFactory.setShareNativeConnection(false);
        stream = stringRedisTemplate.opsForStream();
    }

    /**
     *  向队列中中添加一条消息
     * @param info 消息
     * @return 消息id
     */
    @RequestMapping(method = {RequestMethod.GET},value = "/message/{info}")
    public String sendMessage(@PathVariable String info){
        
        long now = System.currentTimeMillis();
        Gson gson = new GsonBuilder().create();
        
        for(int i = 0;i < 1;i++) {
        	Map<String,Object> map = new LinkedHashMap<>();
        	Map<String,Object> map2 = new LinkedHashMap<>();
        	String[] ar = {"a","b"};
	        map.put("key",i);
	        map2.put("ar", ar);
	        map.put("data",map2);
	        final String v = info + i;
	        
	        Set<String> userIds = new HashSet<>();
            userIds.add("1");
            userIds.add("2");
	        
	        HashMap<String, Object> bodyMap = new HashMap<>();
            bodyMap.put("body", "aaa");
            bodyMap.put("type", "fire");
            bodyMap.put("modules", new String[]{"xjwbWebsocket"});

            PropellingInfo propellingInfo = new PropellingInfo();
            propellingInfo.setPropellType("WEBSOCKET");
            propellingInfo.setWebsocketUserIds(userIds);
            propellingInfo.setWebsocketContent(gson.toJson(bodyMap));
            Map mqMap = gson.fromJson(gson.toJson(propellingInfo),Map.class);
            mqMap.put("data", gson.toJson(propellingInfo));
            redisCacheUtil.setCacheObject("todd", propellingInfo);
            System.out.println(redisCacheUtil.get("todd",PropellingInfo.class));
        	executor.execute(new Runnable() {
				@Override
				public void run() {
					String key = RedisConstants.STREAM_KEY;
					String annoKey = RedisConstants.FIRE_STREAM_KEY;
					String sKey = RedisConstants.STREAM_KEY3;
					String fKey = RedisConstants.FURY_STREAM_KEY;
					String nsKey = RedisConstants.NS_STREAM_KEY;
					
					String recordId = commonService.addStreamMap(key, mqMap);
                    recordId = commonService.addStreamMap(key + "3", mqMap);
                    commonService.addStreamObj(fKey, mqMap);
//					recordId = commonService.addStreamMap(annoKey, mqMap);
//					recordId = commonService.addStreamObj(nsKey, propellingInfo);
//					String recordId = commonService.addStream(annoKey, map);
//					String recordId = commonService.addStream(sKey, v);
				}
        		
        	});
        }
//        executor.shutdown();
        log.info("use {} senconds",(System.currentTimeMillis() - now)/1000);
//        Map<String,String> map1 = new LinkedHashMap<>();
//        map1.put("key1",info);
//        RecordId recordId1 = stream.add("cjfather:queue:1", map1);
        return "ok";
    }
    /**
     *  获取该队列信息
     * @param key 队列key
     * @return 消息id
     */
    @RequestMapping(method = {RequestMethod.GET},value = "/group/{key}")
    public String getMessageGroup(@PathVariable String key){
        StreamInfo.XInfoGroups groups = stream.groups(key);
        StreamInfo.XInfoStream info = stream.info(key);
        log.info("groups:{}",groups.toString());
        log.info("info:{}",info.toString());
        return "ok";
    }

    /**
     *  在对应的队列中创建组
     * @param key 队列key
     * @param group 组名
     * @return 返回ok
     */
    @RequestMapping(method = {RequestMethod.GET},value = "/group/{key}/{group}")
    public String createGroup(@PathVariable String key,@PathVariable String group){
        log.info("在{}下创建了组{}",key,group);
        return stream.createGroup(key,ReadOffset.from("0-0"), group);
    }

    /**
     * 通key和消费组获取所有未被ack的信息  {注意：消费者不能为数字、如果是数字这块会出现转异常}
     * @param key  队列key
     * @param group 组名
     */
    @RequestMapping(method = {RequestMethod.GET},value = "/pending/{key}/{group}")
    public String getPending(@PathVariable String key,@PathVariable String group){
        PendingMessagesSummary pending = stream.pending(key, group);
        // 所有pending消息的数量
        Long totalPendingMessages = pending.getTotalPendingMessages();
        // 消费组名称
        String groupName= pending.getGroupName();
        // pending队列中的最小ID
        String minMessageId = pending.minMessageId();

        // pending队列中的最大ID
        String maxMessageId = pending.maxMessageId();

        log.info("消费组：{}，一共有{}条pending消息，最大ID={}，最小ID={}", groupName, totalPendingMessages, minMessageId, maxMessageId);

        // 查询每个消费者pending数量
        Map<String, Long> pendingMessagesPerConsumer = pending.getPendingMessagesPerConsumer();
        // 消费者名称及待消费的数据条数
        log.info(String.valueOf(pendingMessagesPerConsumer));

        return "ok";
    }

    /**
     * 获取未被ack的信息
     * @param key 队列key
     * @param group 组名
     * @param groupName 消费者
     */
    @RequestMapping(method = {RequestMethod.GET},value = "/read/{key}/{group}/{groupName}")
    public String getRead(@PathVariable String key,@PathVariable String group,@PathVariable String groupName){
        // 获取key所有的信息{堵塞状态}
        List<MapRecord<String, String, String>> retVal = stream.read(StreamReadOptions.empty().block(Duration.ofMillis(5000L)), StreamOffset.fromStart(key));
        // 获取key对应组对应消费者待ack的信息
//        List<MapRecord<String, String, String>> retVal = stream.read(Consumer.from(group, groupName), StreamOffset.create(key,ReadOffset.lastConsumed()));
        // 遍历消息
        for (MapRecord<String, String, String> record : retVal ) {
            // 消费消息
            log.info("消息id={}, 消息value={}", record.getId(), record.getValue());
            // 手动ack消息
            stream.acknowledge(group, record);
        }
        return "ok";
    }

    /**
     *  手动标记队列为ack消费状态从pending移除
     * @param key  队列key
     * @param group 组名
     * @param recordId 消息id
     * @return 消费条数
     */
    @RequestMapping(method = {RequestMethod.GET},value = "/ack/{key}/{group}/{recordId}")
    public Long ack(@PathVariable String key,@PathVariable String group,@PathVariable String recordId){
        Long acknowledge = stream.acknowledge(key, group, recordId);
        return acknowledge;
    }
    
    @RequestMapping(method = {RequestMethod.GET},value = "/list/add/{msg}")
    public List add(@PathVariable String msg){
    	redisCacheUtil.setCacheList("list:test", msg);
        return redisCacheUtil.getCacheList("list:test");
    }

    @RequestMapping("/list/del/{msg}")
    public List del(@PathVariable String msg){
    	long size = redisCacheUtil.delCacheList("list:test", msg);
    	List list = redisCacheUtil.getCacheList("list:test");
    	redisCacheUtil.delCacheList("list:test", "3");
    	redisCacheUtil.delCacheList("list:test", 3);
    	list.add(size);
        return list;
    }


}
